#include "io_thread.h"
#include "session.h"
#include "defer.h"
#include "MsgNode.h"
IOThread::IOThread(int index):_event_count(1024), _stop(true), _index(index), _expanded_once(false)
{
    _event_fd = eventfd(0, EFD_NONBLOCK);
    if (_event_fd == -1)
    {
        perror("eventfd");
        std::exit(1);
    }

    _epoll_fd = epoll_create1(0);
    if (_epoll_fd == -1)
    {
        perror("epoll_create1");
        std::exit(1);
    }

    auto add_res = add_fd(_event_fd, EPOLLIN);
    if (!add_res)
    {
        perror("epoll add eventfd failed");
        std::exit(1);
    }

    //为epoll_wait 申请缓冲区
    _event_addr.resize(_event_count);
}

bool IOThread::add_fd(int fd, int events){
    struct epoll_event ev2{};
    ev2.events = events;
    ev2.data.fd = fd;
    if(epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, &ev2) == 0){
        return true;
    }
    perror("fd add to epoll failed, try to modify");
    // 尝试修改已存在的 fd 的事件（改用 EPOLL_CTL_MOD）
    if (epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, fd, &ev2) == -1)
    {
        perror("fd modify failed: ");
        return false;
    }
    return true;
}

int IOThread::set_nonblocking(int fd)
{
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags == -1)
        return -1;
    return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

bool IOThread::mod_fd(int fd, int events)
{
    struct epoll_event ev2
    {
    };

    ev2.events = events;
    ev2.data.fd = fd;

    if (epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, fd, &ev2) == -1)
    {
        perror("fd modify failed: ");
        return false;
    }

    return true;
}

bool IOThread::del_fd(int fd)
{
    auto it = _sessions.find(fd);
    if (it == _sessions.end()) {
        return false;
    }
    std::cout << "Closing and removing session for fd=" << fd << std::endl;
    if (epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, NULL) == -1)
    {
        perror("fd del failed: ");
        return false;
    }
    close(fd);
    _sessions.erase(it);
    std::cout << "Session for fd=" << fd << " has been removed." << std::endl;
    return true;
}

IOThread::~IOThread()
{
    std::cout << "IOThread 【" << _index << "】exit" << std::endl;
}

void IOThread::start(){
    _stop = false;
    _thread = std::thread([this]{
        this->loop();
    });
}

void IOThread::enqueue_new_con(int fd){
    auto task = std::make_shared<IOTask>(fd, TaskType::RegisterConn);
    enqueue_task(task);
}

void IOThread::catche_new_con(int fd){
    auto task = std::make_shared<IOTask>(fd, TaskType::RegisterConn);
    {
        std::lock_guard<std::mutex> lk(_task_mtx);
        _tasks.push(task);
    }
}

void IOThread::wakeup(){
    uint64_t one = 1;
    auto n = write(_event_fd, &one, sizeof(one));
}

void IOThread::stop()
{
    _stop = true;
    auto task = std::make_shared<IOTask>(_event_fd, TaskType::Shutdown);
    enqueue_task(task);
}


void IOThread::join()
{
    if (_thread.joinable())
    {
        _thread.join();
    }

    std::cout << "IOThread join exit" << std::endl;
}

void IOThread::enqueue_task(std::shared_ptr<IOTask> task)
{
    {
        std::lock_guard<std::mutex> lk(_task_mtx);
        _tasks.push(task);
    }
    wakeup();
}

void IOThread::enqueue_send_data(int fd, const std::string &msg, int msgtype)
{
    auto task = std::make_shared<IOTask>(fd, TaskType::SendData, msg, msgtype);
    enqueue_task(task);
}


void IOThread::clear_fd(int fd)
{
    _sessions.erase(fd);
    del_fd(fd);
    close(fd);
}

bool IOThread::deal_enque_tasks(){
    std::queue<std::shared_ptr<IOTask>> q;
    {
        std::lock_guard<std::mutex> lk(_task_mtx);
        std::swap(q, _tasks);
        /*
        减少锁的持有时间之后处理任务时无需持有锁（任务已在临时队列 q 中），其他线程可以立即向 _tasks 中添加新任务，减少阻塞。
        交换后，所有待处理任务都在临时队列 q 中，后续可以在无锁的环境下遍历 q 处理任务，避免了处理每个任务时都需要加锁解锁的开销。
        */
    }
    while(!q.empty()){
        auto task = q.front();
        q.pop();
        if(task->_type == TaskType::RegisterConn){
            auto sess = std::make_shared<Session>(task->_fd, shared_from_this());
            _sessions[task->_fd] = sess;
            set_nonblocking(task->_fd);
            add_fd(task->_fd, EPOLLIN | EPOLLET);
            continue;
        }
        if(task->_type == TaskType::SendData){
            auto iter = _sessions.find(task->_fd);
            if(iter == _sessions.end()){
                continue;
            }
            auto data_buf = std::make_shared<DataBuf>(static_cast<uint16_t>(task->_msgtype), task->_data);
            
            //数据正在发送，则直接放入队列
            if (iter->second->_send_stage == SendStage::SENDING)
            {
                iter->second->SendData(data_buf);
                continue;
            }

            //没有数据发送，则直接发送
            auto send_res = iter->second->SendData(data_buf);
            if (send_res == -1)
            {
                _sessions.erase(task->_fd);
                del_fd(task->_fd);
                close(task->_fd);
                continue;
            }

            //遇到EAGAIN，放入EPOLL监听
            if (send_res == 1)
            {
                mod_fd(task->_fd, EPOLLET | EPOLLIN | EPOLLOUT);
                continue;
            }
            //发送正常则跳过后面内容
            continue;
        }
        if (task->_type == TaskType::Shutdown)
        {
            return false;
        }
    }
    return true;
}

void IOThread::handle_epollout(std::shared_ptr<Session> sess)
{
    if(sess->_send_stage == SendStage::SENDING) return;
    //RAII defer析构执行函数
    sess->_send_stage = SendStage::SENDING;
    Defer defer([sess]() {
        sess->_send_stage = SendStage::NO_SEND;
    });
    auto send_res = sess->SendData(nullptr);
    if(send_res == IOStatus::IO_ERROR){
        clear_fd(sess->GetFd());
    }
    //所有数据发送完成，取消EPOLLOUT
    if(send_res == IOStatus::IO_SUCCESS){
        mod_fd(sess->GetFd(), EPOLLET | EPOLLIN);
    }
    // struct epoll_event ev;
    // ev.events = EPOLLIN | EPOLLET;
    // ev.data.fd = sess->_fd;
    // epoll_ctl(_epoll_fd, EPOLL_CTL_MOD, sess->_fd, &ev);
}

void IOThread::loop() {
    while (!_stop) {
        int nfds = epoll_wait(_epoll_fd, _event_addr.data(), (int)_event_count, HEARTBEAT_TIMEOUT);
        if (nfds < 0) {
            if (errno == EINTR)
                continue; // 被信号打断，重试epoll_wait
            perror("epoll_wait");
            break;
        }

        // 如果命中上限，动态扩容一倍
        if ((size_t)nfds == _event_count) {
            _event_count *= 2;
            _event_addr.resize(_event_count);
        }

        for (int i = 0; i < nfds; i++) {
            int fd = _event_addr[i].data.fd;
            uint32_t evs = _event_addr[i].events;

            // 处理错误事件
            if (evs & (EPOLLERR | EPOLLHUP)) {
                int err = 0;
                socklen_t errlen = sizeof(err);
                getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen);
                fprintf(stderr, "fd=%d error: %s\n", fd, strerror(err));
                clear_fd(fd);
                continue;
            }

            // 处理事件通知（任务队列）
            if (fd == _event_fd) {
                uint64_t cnt;
                read(_event_fd, &cnt, sizeof(cnt));

                auto deal_res = deal_enque_tasks();
                if (!deal_res) {
                    std::cout << "io_thread receive exit eventfd" << std::endl;
                    return;
                }
                continue;
            }

            // 处理读写事件
            if (evs & (EPOLLIN | EPOLLOUT)) {
                auto iter = _sessions.find(fd);
                if (iter == _sessions.end()) {
                    continue;
                }
                auto sess = iter->second;

                // 处理读事件
                if (evs & EPOLLIN) {
                    // 检查会话指针有效性
                    if (!sess) {
                        std::cerr << "Invalid session pointer for fd: " << fd << std::endl;
                        clear_fd(fd);
                        continue;
                    }
                    
                    try {
                        IOStatus recv_status = sess->RecvData();
                        // 处理所有可能的返回状态
                        if (recv_status == IOStatus::IO_ERROR) {
                            std::cerr << "Receive error for fd: " << fd << std::endl;
                            clear_fd(fd);
                        } else if (recv_status != IOStatus::IO_SUCCESS) {
                            // 处理其他非成功状态（如部分接收等）
                            std::cerr << "Unexpected receive status " << static_cast<int>(recv_status) 
                                    << " for fd: " << fd << std::endl;
                            // 根据实际情况决定是否关闭连接或其他处理
                        }
                    } catch (const std::exception& e) {
                        std::cerr << "Exception occurred while receiving data for fd: " << fd 
                                << ", error: " << e.what() << std::endl;
                        clear_fd(fd);
                    } catch (...) {
                        std::cerr << "Unknown exception occurred while receiving data for fd: " << fd << std::endl;
                        clear_fd(fd);
                    }
                }

                // 处理写事件
                if (evs & EPOLLOUT) {
                    handle_epollout(sess);
                }

                continue;
            }
        }
        check_timeouts();// 在处理完所有网络事件后，或 epoll_wait 超时返回后，执行一次超时检查
    }
}
void IOThread::check_timeouts(){
    auto now = std::chrono::steady_clock::now();
    std::vector<int>timeout_fds;
    for(const auto& pair : _sessions){
        auto& session = pair.second;
        auto duration = std::chrono::duration_cast<std::chrono::seconds>(now - session->get_last_active_time());
        if(duration.count() > HEARTBEAT_TIMEOUT){
            std::cout << "Session[" <<session->GetFd() <<"] time out. Closing." << std::endl;
            timeout_fds.push_back(session->GetFd());
        }
    }

    for(int fd : timeout_fds){
        del_fd(fd);
    }
}
